package pCloud.service;

import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Iterator;

import pCloud.message.ClientRequest;
import pCloud.message.ClientResponse;
import pCloud.message.DatabaseReply;
import pCloud.message.Message;
import pCloud.message.NumberMessage;
import pCloud.message.PolicyManagementRequest;
import pCloud.message.PurgeCacheReply;
import pCloud.message.PurgeCacheRequest;
import pCloud.message.ReplyMessage;
import pCloud.message.RequestFile;
import pCloud.message.RequestedResource;
import pCloud.transport.Communicator;
import pCloud.transport.SocketCommunicator;
import pCloud.transport.SocketTransportHandle;
import pCloud.transport.TransportHandle;

/**
 * Handle request coming from the Client
 */
public class ProxyServerClientProtHandle extends ProtocolHandle {

	ProxyHelper proxyHelper;
	int clientRequestId = 0;
	CachingHandle cacheHandler = null;
	IncompleteResponse incompleteResponses;

	public void init(TransportHandle conn, Communicator comm) {
		super.init(conn, comm);
		this.proxyHelper = new ProxyHelper();
		this.clientRequestId = 0;
	}

	public void initializeCacheHandler(CachingHandle c) {
		this.cacheHandler = c;
	}

	public void initializeIncompleteResponse(IncompleteResponse ir) {
		this.incompleteResponses = ir;
	}

	@Override
	public void receive(Object o) throws Exception {
		this.handleClientRequest((Message) o);
	}

	@Override
	public void send(Object m) throws Exception {
		conn.send(m);
	}

	private void handleClientRequest(Message m) throws Exception {
		if (m.getMessageType() != Message.CLIENT_TO_PROXY_MSG) {
			System.err.println("CLIENT_TO_PROXY msg expected, got "
					+ m.getMessageType());
			return;
		}

		this.communicator.getLogger().debug(
				"CLIENT_TO_PROXY msg received from " + m.getFrom());

		// 1. Unwrap the message
		Object content = m.getMessageContent();
		InetSocketAddress clientAddress = m.getFrom();
		InetSocketAddress dataServerAddrss = m.getTo();

		if (content instanceof ClientRequest) {
			// assign a global, unique ID for this request
			((ClientRequest) content).setRequestId(this.clientRequestId++);

			ArrayList<Message> messages = this.processClientRequest(
					(ClientRequest) content, clientAddress);
			this.communicator.getLogger().debug(
					"Parsing request from the client, " + content
							+ ". Number of messages to be sent is "
							+ messages.size());
			Message wrappedMsg;
			InetSocketAddress localSock;
			for (int i = 0; i < messages.size(); i++) {
				localSock = (InetSocketAddress) this.communicator
						.getConnection(messages.get(i).getTo())
						.getLocalAddress();
				wrappedMsg = new Message(localSock, messages.get(i).getTo(),
						messages.get(i), Message.PROXY_TO_DATA_MSG);
				this.communicator.send(wrappedMsg.getTo(), wrappedMsg);
			}

			this.waitForRepliesThenSend((ClientRequest) content, clientAddress);
		} else if (content instanceof PurgeCacheRequest) {// if it is a request
															// to clear the
															// cache
			this.cacheHandler.purgeCache();
			this.communicator.send(m.getFrom(), new Message(null, m.getFrom(),
					new PurgeCacheReply(), Message.PROXY_TO_CLIEN_MSG));
			//then close
			this.communicator.getConnection(m.getFrom()).close();
		} else { // just forward the message along
					// purging the cache for adding/removing policies.
			InetSocketAddress serverAddress = m.getTo();
			if (content instanceof PolicyManagementRequest) {
				PolicyManagementRequest pmr = (PolicyManagementRequest) content;
				if (pmr.getType() == PolicyManagementRequest.ADD
						|| pmr.getType() == PolicyManagementRequest.REMOVE) {
					if (this.cacheHandler != null)
						this.cacheHandler.purgeCache();
					this.communicator.getLogger().debug(
							"Purging cache, because of policy update");
				}

				serverAddress = this.communicator.getDataIdMap()
						.get(pmr.getDataId()).getAddress();
			}
			Message wrappedMessage = new Message(
					(InetSocketAddress) this.communicator.getConnection(
							serverAddress).getLocalAddress(), serverAddress, m,
					Message.PROXY_TO_DATA_MSG);
			this.communicator.send(wrappedMessage.getTo(), wrappedMessage);
		}

	}

	/**
	 * Parsing the ClientRequest message and constructing a list of messages to
	 * be sent to different servers.
	 * 
	 * Assume that there is at most one message constructed for each data server
	 * 
	 * @throws URISyntaxException
	 */
	private ArrayList<Message> processClientRequest(ClientRequest request,
			InetSocketAddress clientAddress) throws URISyntaxException {
		ArrayList<Message> messages = new ArrayList<Message>();

		ArrayList<RequestedResource> resources = request.getResources();
		Message msgToServer = null;

		ClientResponse response = new ClientResponse(request.getRequestId(),
				resources.size());

		this.incompleteResponses.addResponse(response);
		RequestedResource rr;
		String dataId = null;
		for (int i = 0; i < resources.size(); i++) {
			rr = resources.get(i);
			dataId = rr.getDataId();

			// translate the dataID into database ID
			rr.setDatabaseId(communicator.getDataIdMap().get(dataId)
					.getDatabaseName());
			rr.setHost(communicator.getDataIdMap().get(dataId).getAddress());

			RequestFile rf = (RequestFile) proxyHelper.createRequest(
					request.getRequestId(), rr.getJoiningColumn(),
					rr.getName(), rr.getRole(), rr);

			// Check for cached values
			ReplyMessage cachedValue = null;
			if (this.cacheHandler != null)
				cachedValue = (ReplyMessage) this.cacheHandler
						.getCachedValue(rf);

			if (cachedValue == null) {
				if (this.cacheHandler != null)
					((DataCaching) this.cacheHandler).addToTempMap(rf,
							rr.getHost());
				// System.out.println("CACHE MISS");
				msgToServer = new Message(clientAddress, rr.getHost(), rf,
						Message.CLIENT_TO_PROXY_MSG);
				messages.add(msgToServer);
			} else {// if there is a cached value
				synchronized (response) {
					response.addReply(rr.getHost(), cachedValue);
					response.notifyAll();
				}
				// System.out.println("CACHE HIT");
			}
		}
		return messages;
	}

	private void waitForRepliesThenSend(ClientRequest request,
			InetSocketAddress clientAddress) throws Exception {
		SocketCommunicator socketComm = (SocketCommunicator) this.communicator;
		ClientResponse response = this.incompleteResponses
				.getIncompleteResponse(request.getRequestId());

		this.communicator.getLogger().debug(
				"Waiting for replies from data servers");
		synchronized (response) {
			while (response.getNumberOfReplies() != 0)
				try {
					response.wait();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
		}
		// after this, all replies from different data servers have arrived at
		// the proxy.
		this.communicator.getLogger().debug(
				"Got all the replies from data server");
		this.incompleteResponses.removeResponse(request.getRequestId());

		if (this.cacheHandler != null)
			System.out.println("CACHE_HIT "
					+ this.cacheHandler.getCacheHitRate());
		else
			System.out.println("CACHE_HIT = 0.0");
		// output cache performance
		if (this.cacheHandler != null)
			this.communicator.getLogger().debug(
					"CACHE HIT = " + this.cacheHandler.getCacheHitRate());

		// 1. Check if there's a need to join, i.e. the joiningColumns are not
		// null and not empty
		if (request.getJoinColumns() != null
				&& request.getJoinColumns().size() != 0) {
			this.communicator.getLogger().debug(
					"Merging data from different sources");
			ReplyMessage replyMsg = this.proxyHelper.mergeResponses(request,
					response);
			Message msgToClient = new Message(null, clientAddress, replyMsg,
					Message.PROXY_TO_CLIEN_MSG);			
			this.communicator.send(msgToClient.getTo(), msgToClient);
			this.communicator.removeConnection(this);
			this.conn.close();
		} else {// forwarding the response back
			this.communicator.getLogger().debug(
					"Forwarding data back to the client, without merging, join columns "
							+ request.getJoinColumns() + " of size "
							+ request.getJoinColumns().size());
			Iterator<InetSocketAddress> it = response.getReplies().keySet()
					.iterator();
			ReplyMessage replyMsg;
			Message msgToClient;
			InetSocketAddress fromAddress = null;
			while (it.hasNext()) {
				fromAddress = it.next();
				replyMsg = response.getReplies().get(fromAddress);
				msgToClient = new Message(fromAddress, clientAddress, replyMsg,
						Message.PROXY_TO_CLIEN_MSG);
				this.communicator.send(msgToClient.getTo(), msgToClient);
			}
			this.communicator.removeConnection(this);
			this.conn.close();
		}
	}
}
